package com.udf.flink.scala.apitest.trigger


import java.time.Duration

import com.udf.flink.scala.apitest.checkpoint.Obj1
import com.udf.flink.scala.apitest.window.MinDataReduceFunction
import org.apache.flink.api.common.eventtime.{SerializableTimestampAssigner, WatermarkStrategy}
import org.apache.flink.api.scala.createTypeInformation
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.timestamps.AscendingTimestampExtractor
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.triggers.EventTimeTrigger

object TestEventTimeTrigger extends App {
  val environment:StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
//  environment.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
  val stream1: DataStream[String] = environment.socketTextStream("localhost",9999)
  stream1.print("stream1")
  val stream2: DataStream[Obj1] = stream1.map(data => {
    val arr = data.split(",")
    //println(arr)
    Obj1(arr(0), arr(1), arr(2).toLong)
  }).assignTimestampsAndWatermarks(WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofSeconds(10))
    .withTimestampAssigner(new SerializableTimestampAssigner[Obj1] {
      override def extractTimestamp(element: Obj1, recordTimestamp: Long): Long = element.time * 1000
    }))
  stream2.keyBy(_.id)
    .window(TumblingEventTimeWindows.of(Time.seconds(6)))
    .trigger(EventTimeTrigger.create()) // 设置 EventTimeTrigger
    .reduce(new MinDataReduceFunction)
    .print("TumblingEventTimeWindow")
  environment.execute()
}

/*
WindowedStream有几个参数，其中构造器要求的是input及windowAssigner参数，然后还有Trigger、Evictor、allowedLateness、OutputTag这几个可选参数；
另外还必须设置operation function，主要有ReduceFunction、AggregateFunction、FoldFunction(废弃)、ProcessWindowFunction这几个
windowAssigner主要用来决定元素如何划分到window中，
这里主要有TumblingEventTimeWindows/TumblingProcessingTimeWindows、
SlidingEventTimeWindows/SlidingProcessingTimeWindows、
EventTimeSessionWindows/ProcessingTimeSessionWindows、GlobalWindows这几个
Trigger用来触发window的发射，Evictor用来在发射window的时候剔除元素，allowedLateness用于指定允许元素落后于watermark的最大时间，
超出则被丢弃(仅仅对于event-time window有效)，OutputTag用于将late数据输出到side output，
可以通过SingleOutputStreamOperator.getSideOutput(OutputTag)方法来获取

 GlobalWindows继承了WindowAssigner，key类型为Object，窗口类型为GlobalWindow；
 GlobalWindow继承了Window，它的maxTimestamp方法与TimeWindow不同，TimeWindow有start和end属性，其maxTimestamp方法返回的是end-1；
 而GlobalWindow的maxTimestamp方法返回的是Long.MAX_VALUE；GlobalWindow定义了自己的Serializer
GlobalWindows的assignWindows方法返回的是GlobalWindow；getDefaultTrigger方法返回的是NeverTrigger；
getWindowSerializer返回的是GlobalWindow.Serializer()；isEventTime返回的为false
NeverTrigger继承了Trigger，其onElement、onProcessingTime、onProcessingTime返回的TriggerResult均为TriggerResult.CONTINUE；
该行为就是不做任何触发操作；如果需要触发操作，则需要在定义window操作时设置自定义的trigger，覆盖GlobalWindows默认的NeverTrigger
stream
       .keyBy(...)               // keyedStream上使用window
       .window(...)              // 必选: 指定窗口分配器( window assigner)
      [.trigger(...)]            // 可选: 指定触发器(trigger),如果不指定，则使用默认值
      [.evictor(...)]            // 可选: 指定清除器(evictor),如果不指定，则没有
      [.allowedLateness(...)]    // 可选: 指定是否延迟处理数据，如果不指定，默认使用0
      [.sideOutputLateData(...)] // 可选: 配置side output，如果不指定，则没有
       .reduce/aggregate/fold/apply() // 必选: 指定窗口计算函数
      [.getSideOutput(...)]      // 可选: 从side output中获取数据

 */